package com.IntelligentCommunity.consumer;

import com.IntelligentCommunity.algorithm.api.AlgorithmRemote;
import com.IntelligentCommunity.algorithm.domain.FaceRecognitionConf;
import com.IntelligentCommunity.common.core.domain.R;
import com.IntelligentCommunity.common.core.utils.file.Url2Base64;
import com.IntelligentCommunity.community.api.RemotePopulationService;
import com.IntelligentCommunity.deploy.api.DeployAlarmRemote;
import com.IntelligentCommunity.deploy.api.WebSocketRemote;
import com.IntelligentCommunity.deploy.domain.DeployAlarm;
import com.IntelligentCommunity.es.api.RemoteEsMatchingDataService;
import com.IntelligentCommunity.es.domain.vo.EsDocInsertVo;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static com.IntelligentCommunity.common.core.constant.KafkaNames.TOPIC_NAME_FACE_BASE64;
import static com.alibaba.fastjson.JSONPatch.OperationType.test;

/**
 * @Author：ZHP
 * @Package：com.IntelligentCommunity.kafka.consumer
 * @Project：IntelligentCommunity
 * @name：DataAccessConsumer
 * @Date：2024/7/19 10:25`
 * @Filename：DataAccessConsumer
 */
@Component
@Log4j2
public class DataAccessConsumer {

    @Autowired
    private RemotePopulationService populationService;

    @Autowired
    private RemoteEsMatchingDataService esMatchingDataService;

    @Autowired
    private AlgorithmRemote algorithmRemote;
    @Autowired
    private   WebSocketRemote webSocketRemote;

    @Autowired
    private DeployAlarmRemote deployAlarmRemote;

    @KafkaListener(topics = TOPIC_NAME_FACE_BASE64,
            groupId = "group11",
            errorHandler = "myKafkaListenerErrorHandler",
            containerFactory = "kafkaListenerContainerFactory")
    public void faceCapture(ConsumerRecord<String ,Object> record, Acknowledgment ack){
        log.info("接收到消息1：" + record.value());
        Object value = record.value();
        Map map = JSON.parseObject(JSON.toJSONString(value), Map.class);

        // 调用远程服务将数据保存到ES
        EsDocInsertVo esDocInsertVo = new EsDocInsertVo();
        // 设置数据
        esDocInsertVo.setData(map);
        // 设置索引名称
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMM");
        // 获取当前日期
        String format = sdf.format(new Date());
        // 设置索引名称
        esDocInsertVo.setIndexName("face" + format);
        // 创建一个包含单个EsDocInsertVo对象的ArrayList
        ArrayList<EsDocInsertVo> esDocInsertVos = new ArrayList<>();
        // 将EsDocInsertVo对象添加到ArrayList中
        esDocInsertVos.add(esDocInsertVo);
        // 标记为人脸抓拍
        esDocInsertVo.setFlag(1);
        // 调用远程服务将数据保存到ES
        esMatchingDataService.insertCaptureData(esDocInsertVos);
        webSocketRemote.sendWran("123456");
        // 确认消息已处理
        ack.acknowledge();

    }

    /**
     * 人脸识别
     * @param record 消息
     * @param ack 确认消息
     */

    @KafkaListener(topics = TOPIC_NAME_FACE_BASE64,
            groupId = "group2",
            errorHandler = "myKafkaListenerErrorHandler",
            containerFactory = "kafkaListenerContainerFactory")
    public void faceRec(ConsumerRecord<String ,Object> record, Acknowledgment ack){
        log.info("接收到消息2：" + record.value());
        Object value = record.value();
        Map hashMap = JSON.parseObject(JSON.toJSONString(value), Map.class);
        String image = (String) hashMap.get("image");
        FaceRecognitionConf recognitionConf = new FaceRecognitionConf();
        recognitionConf.setImg(image);
        recognitionConf.setFlag(0);

        R r = algorithmRemote.faceRec(recognitionConf);
        Map<String, Object> data = (Map) r.getData();

        // 检查结果是否包含"result"字段
        if (data.containsKey("result")) {
            Object result = data.get("result");

            // 将结果转换为JSON字符串
            String jsonString = JSONObject.toJSONString(result);
            // 解析JSON字符串为List<Map>
            List<Map> maps = JSONObject.parseArray(jsonString, Map.class);
            Map map = maps.get(0);
            Object subjects = map.get("subjects");
            // 将结果转换为JSON字符串
            String subjectsJsonString = JSONObject.toJSONString(subjects);
            // 解析JSON字符串为List<Map>
            List<Map> subjectsMaps = JSONObject.parseArray(subjectsJsonString, Map.class);
            Map subjectMap = subjectsMaps.get(0);
            Double similarity = ((Number) subjectMap.get("similarity")).doubleValue();
            if (similarity < 0.8) {
                return;
            }
            String subject = String.valueOf(subjectMap.get("subject")).split(",")[1];
            populationService.updateTimeById(Long.valueOf(subject));
        }
        // 确认消息已处理
        ack.acknowledge();

    }


    /**
     * 布控人脸识别
     * @param record
     * @param ack
     */
    @KafkaListener(topics = TOPIC_NAME_FACE_BASE64,
            groupId = "group19",
            errorHandler = "myKafkaListenerErrorHandler",
            containerFactory = "kafkaListenerContainerFactory")
    public void deployFaceRec(ConsumerRecord<String ,Object> record, Acknowledgment ack){
        log.info("接收到消息3：" + record.value());
        Object value = record.value();
        Map hashMap = JSON.parseObject(JSON.toJSONString(value), Map.class);
        String image = (String) hashMap.get("image");
        FaceRecognitionConf recognitionConf = new FaceRecognitionConf();
        recognitionConf.setImg(image);
        recognitionConf.setFlag(1);

        R r = algorithmRemote.faceRec(recognitionConf);
        Map<String, Object> data = (Map) r.getData();
        // 检查结果是否包含"result"字段
        if (data.containsKey("result")) {
            Object result = data.get("result");
            // 将结果转换为JSON字符串
            String jsonString = JSONObject.toJSONString(result);
            // 解析JSON字符串为List<Map>
            List<Map> maps = JSONObject.parseArray(jsonString, Map.class);
            Map map = maps.get(0);
            Object subjects = map.get("subjects");
            if (null == subjects){
                return;
            }
            // 将结果转换为JSON字符串
            String subjectsJsonString = JSONObject.toJSONString(subjects);
            // 解析JSON字符串为List<Map>
            List<Map> subjectsMaps = JSONObject.parseArray(subjectsJsonString, Map.class);
            Map subjectMap = subjectsMaps.get(0);
            Double similarity = ((Number) subjectMap.get("similarity")).doubleValue();
            String subject = String.valueOf(subjectMap.get("subject"));
            DeployAlarm deployAlarm = new DeployAlarm();

            deployAlarm.setCaptureTime(new Date());
            deployAlarm.setImage(image);
            deployAlarm.setSimilarity(similarity * 100.0);
            deployAlarm.setSubject(subject);
            deployAlarm.setSecurityFacilitesId(Long.valueOf(String.valueOf(hashMap.get("realSecurityFacilitiesId"))));
            deployAlarm.setType(0L);

            deployAlarmRemote.alarm2mq(deployAlarm);
        }
        // 确认消息已处理
        ack.acknowledge();

    }
}
